package org.swift.mashup.engine.model;

import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

import javax.xml.parsers.ParserConfigurationException;

import org.apache.commons.io.IOUtils;
import org.apache.log4j.Logger;
import org.swift.commons.xml.XmlUtils;
import org.swift.mashup.engine.InitializationException;
import org.swift.mashup.engine.MashupException;
import org.swift.mashup.engine.concurrency.ConcurrencySettingsLoader;
import org.swift.mashup.engine.concurrency.ConcurrentConfiguration;
import org.swift.mashup.engine.concurrency.ConcurrentContext;
import org.swift.mashup.engine.concurrency.LocalConcurrentContext;
import org.swift.mashup.engine.concurrency.RestfulConcurrentContext;
import org.swift.mashup.engine.concurrency.StaticConcurrentConfiguration;
import org.swift.mashup.engine.concurrency.agenda.ActionConfig;
import org.swift.mashup.engine.concurrency.agenda.ActionTemplate;
import org.swift.mashup.engine.concurrency.execution.ExecutionSession;
import org.swift.mashup.engine.concurrency.execution.ExecutionOutput;
import org.swift.mashup.engine.concurrency.execution.Rule;
import org.swift.mashup.engine.concurrency.execution.RuleSet;
import org.w3c.dom.Document;
import org.xml.sax.SAXException;

@SuppressWarnings("unchecked")
public class MashupImpl implements Mashup {
	private static final Logger logger = Logger.getLogger(Mashup.class);
	private static final String WORKING_DIRECTORY_PROP = "org.swift.mashup.engine.concurrency.WorkingDirectory";

	private ActionConfig actionConfig;
	private Document mashupXml;
	private boolean isLoaded = false;
	private String name;
	private ConcurrentConfiguration concurrentCfg;

	public MashupImpl(String name) {
		this.name = name;
	}

	public MashupImpl() {
	}

	public MashupImpl(InputStream stream) throws ParserConfigurationException,
			SAXException, IOException, InitializationException, ParseException {
		Document doc = XmlUtils.toDocument(IOUtils.toString(stream));
		this.load(doc);
	}

	public MashupImpl(Document mashupXml) throws ParserConfigurationException,
			SAXException, IOException, InitializationException, ParseException {
		this.load(mashupXml);
	}

	public MashupImpl(ActionConfig actionConfig) {
		this.actionConfig = actionConfig;
		this.isLoaded = true;
	}

	@Override
    public Collection<String> getParameterNames() {
		return ((MashupXmlActionConfig) actionConfig).getInputParameters()
				.keySet();
	}

	@Override
    public void load(InputStream stream) throws ParserConfigurationException,
			SAXException, IOException, InitializationException, ParseException {
		Document doc = XmlUtils.toDocument(IOUtils.toString(stream));
		this.load(doc);
	}

	@Override
    public void load(Document mashupXml) throws ParserConfigurationException,
			SAXException, IOException, InitializationException, ParseException {
		this.mashupXml = mashupXml;
		try {
			if (this.actionConfig == null){
				this.actionConfig = new MashupXmlActionConfig();
			}
			((MashupXmlActionConfig) this.actionConfig).load(mashupXml,name);
			this.isLoaded = true;
		} catch (InitializationException e) {
			throw e;
		}
	}

	@Override
    public ExecutionOutput run(Map parameters) {
		if (!this.isLoaded) {
			if (this.mashupXml != null) {
				try {
					this.load(this.mashupXml);
				} catch (Exception e) {
					// that was the last chance to load.
					throw new RuntimeException(e);
				}
			}
		}
		long start = System.currentTimeMillis();
		ExecutionSession session = null;
		if (!this.isParallel()) {
			session = new ExecutionSession(this.actionConfig.getRuleSet(), this.actionConfig);
		} else {
			try {
				ConcurrentContext ctx = createConcurrentContext();
				session = new ExecutionSession(this.actionConfig.getRuleSet(),
						this.actionConfig, ctx);
			} catch (MashupException me){
				logger.error(String.format("Failed to initialize the concurrent context for '%s'. Falling back to local processing.", this.getName()),me);
				session = new ExecutionSession(this.actionConfig.getRuleSet(), this.actionConfig);
			}
		}
		ExecutionOutput results = session.run(parameters);
		logger.debug("Mashup executed in "
				+ (System.currentTimeMillis() - start) + "ms.");
		return results;
	}
	
	/**
	 * This is exposed as a web method and not meant to be invoked manually.
	 * It uses a local concurrent context by nature because it will not be
	 * calling out to other nodes in the cloud.
	 * 
	 * @param processid
	 * @param sessionid
	 * @param workingDirectory
	 * @param startIndex
	 * @param endIndex
	 * @return
	 * @throws org.swift.mashup.engine.MashupException
	 */
	@Override
    public String runConcurrent(String processid, String sessionid,
                                String workingDirectory, int startIndex, int endIndex) throws MashupException {
		ConcurrentConfiguration config = StaticConcurrentConfiguration.newConfiguration(1, workingDirectory);
		ConcurrentContext context = new LocalConcurrentContext(
				sessionid, config,this.actionConfig.getRuleSet());
		context.createJobSet(processid, startIndex, endIndex);
		context.runJobs();
		return "OK";
	}

	@Override
    public boolean isParallel() {
		return this.actionConfig.isConcurrent();
	}
	
	@Override
    public void setParallel(boolean parallel) {
		this.actionConfig.setConcurrent(parallel);
	}
		
	protected ConcurrentContext createConcurrentContext() throws MashupException {
		if (concurrentCfg == null){
			synchronized (this){
				if (concurrentCfg == null){
					String workingDirectory = System.getProperty(WORKING_DIRECTORY_PROP);
					if (workingDirectory == null){
						logger.warn("No working directory property located [sys prop=" + WORKING_DIRECTORY_PROP + "]");
						workingDirectory = "";
					} else if (logger.isDebugEnabled()) {
						logger.debug("Using working directory " + workingDirectory);
					}
					concurrentCfg = new ConcurrencySettingsLoader(workingDirectory);
				}
			}
		}
		String contextType = concurrentCfg.getContextType();
		if (contextType != null) {
			contextType=contextType.toLowerCase();
		}
		if ("rest".equals(contextType)) {
			return new RestfulConcurrentContext(
					UUID.randomUUID().toString(), concurrentCfg,this.name);
		} else if ("grid".equals(contextType)) {
			return new RestfulConcurrentContext(
					UUID.randomUUID().toString(), concurrentCfg,this.name);
		} else {
			// should use local in file
			return new LocalConcurrentContext(UUID.randomUUID().toString(),
					concurrentCfg,this.getRuleSet());
		}
	}
	
	
	@Override
    public String getName() {
		return this.name;
	}
	
	@Override
    public RuleSet getRuleSet(){
		return this.actionConfig.getRuleSet();
	}
	
	@Override
    public Map<String, Rule> findReferencedRules(String ruleName) {
		Rule rule = this.getRuleSet().getRules().get(ruleName);
		Map<String, Rule> matchedRules = new HashMap<String, Rule>();
		return findReferenceRules(ruleName, rule, matchedRules);
	}

	private Map<String, Rule> findReferenceRules(
			String ruleName, Rule rule, Map<String, Rule> foundRules) {
		if (foundRules.containsKey(ruleName)) {
			return foundRules;
		}
		
		Map<String, Rule> rules = this.actionConfig.getRuleSet().getRules();		
		foundRules.put(ruleName, rule);
		
		Collection<ActionTemplate> actionTemplates = rule.getActions();
		for (ActionTemplate actionTemplate : actionTemplates) {
			From from = actionTemplate.getFrom();
			final String actionRuleName = from.getSource();
			Rule fromRule = rules.get(actionRuleName);
			if (fromRule != null) {
				findReferenceRules(actionRuleName, fromRule, foundRules); //recursive
			}
			Map<String, With> withs = actionTemplate.getWiths();
			for (With with : withs.values()) {
				final String withFrom = with.getFrom();
				if (withFrom != null) {					
					fromRule = rules.get(withFrom);
					if (fromRule != null){
						findReferenceRules(withFrom, fromRule,foundRules); //recursive
					}
				}				
			}
		}
		return foundRules;
	}

	@Override
    public void setNewRules(Collection<Rule> newRules) {
		RuleSet ruleSet = this.cloneOrCreateEmptyRuleSet();
		for (Rule rule : newRules) {
			ruleSet.addRule(rule);	
		}
		this.actionConfig.setRuleSet(ruleSet);		
	}
	
	private RuleSet cloneOrCreateEmptyRuleSet(){
		RuleSet ruleSet = this.actionConfig.getRuleSet();
		if (ruleSet != null){
			return ruleSet.clone(false);
		}
		return new RuleSet();
	}

	@Override
    public Document getMashupXml() {
		return mashupXml;
	}

	@Override
    public Document getSchema() {
		return this.actionConfig.getSchema();
	}
	
	@Override
    public Document getValidationRules() {
		return this.actionConfig.getValidationRules();
	}

	@Override
    public String getDocumentation() {
		return this.actionConfig.getDocumentation();
	}
}
